/*
 *@Type NormalStore.java
 * @Desc
 * @Author urmsone urmsone@163.com
 * @date 2024/6/13 02:07
 * @version
 */
package service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import model.WALLogger;
import model.command.Command;
import model.command.CommandPos;
import model.command.RmCommand;
import model.command.SetCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import utils.*;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class NormalStore implements Store {

    public static final String RW_MODE = "rw";
    public static final String TYPE = "type";
    private final Logger LOGGER = LoggerFactory.getLogger(NormalStore.class);
    private final String logFormat = "[NormalStore][{}]: {}";
    private final WALLogger walLogger = new WALLogger();
    private static int opt_count = 0;

    // 添加集群同步服务引用
    private ClusterSyncService clusterSyncService;

    /**
     * 数据目录
     */
    private final String dataDir;
    /**
     * 读写锁，支持多线程，并发安全写入
     */
    private final ReadWriteLock indexLock;
    /**
     * 内存表，类似缓存
     */
    private TreeMap<String, Command> memTable;

    private HashMap<String, HashMap<String, Command>> tableMemTable;
    /**
     * hash索引，存的是数据长度和偏移量
     */
    private HashMap<String, CommandPos> index;

    private HashMap<String, HashMap<String, CommandPos>> tableIndex;

    /**
     * 暂存数据的日志句柄
     */
    private RandomAccessFile writerReader;

    /**
     * 持久化阈值
     */
//    private final int storeThreshold;
    public NormalStore(String dataDir) {
        this.dataDir = dataDir;
        this.indexLock = new ReentrantReadWriteLock();
        this.memTable = new TreeMap<String, Command>();
        this.index = new HashMap<>();
        this.tableMemTable = new HashMap<String, HashMap<String, Command>>();
        this.tableIndex = new HashMap<String, HashMap<String, CommandPos>>();

        File file = new File(dataDir);
        if (!file.exists()) {
            LoggerUtil.info(LOGGER, logFormat, "NormalStore", "dataDir isn't exist,creating...");
            file.mkdirs();
        }
        walLogger.clear();
        this.reloadIndex();
    }

    public void setClusterSyncService(ClusterSyncService clusterSyncService) {
        this.clusterSyncService = clusterSyncService;
    }

    public String genFilePath(String table) {
        return this.dataDir + File.separator + table + Const.TABLE;
    }


    public void reloadIndex() {
            List<String> tableNames = ConfigLoader.readTableNames();
            if (tableNames.size() == 0){
                // 不存在表 使用默认表名称
                if (!tableMemTable.containsKey(Const.NAME)){
                    tableMemTable.put(Const.NAME,new HashMap<String, Command>());
                }
                if (!tableIndex.containsKey(Const.NAME)){
                    tableIndex.put(Const.NAME,new HashMap<String, CommandPos>());
                }
                String tableFilePath = this.genFilePath(Const.NAME);
                List<String> tableName = new ArrayList<>();
                tableName.add(Const.NAME);
                ConfigLoader.writeTableNames(tableName);
                readIndexInFile(Const.NAME, tableFilePath);

            }else {
                for (String name : tableNames) {
                    if (!tableMemTable.containsKey(name)){
                        tableMemTable.put(name,new HashMap<String, Command>());
                    }
                    if (!tableIndex.containsKey(name)){
                        tableIndex.put(name,new HashMap<String, CommandPos>());
                    }

                    String tableFilePath = this.genFilePath(name);
                    readIndexInFile(name, tableFilePath);
                }
            }

    }
    
    private void readIndexInFile(String tableName, String tableFilePath) {
        try {

            RandomAccessFile file = new RandomAccessFile(tableFilePath, RW_MODE);
            long len = file.length();
            long start = 0;
            //临时存储index的局部变量
            Map<String, CommandPos> newIndex = new HashMap<>();
            file.seek(start);
            while (start < len) {
                int cmdLen = file.readInt();
                byte[] bytes = new byte[cmdLen];
                file.read(bytes);
                JSONObject value = JSON.parseObject(new String(bytes, StandardCharsets.UTF_8));
                Command command = CommandUtil.jsonToCommand(value);
                start += 4;
                if (command != null) {
                    CommandPos cmdPos = new CommandPos((int) start, cmdLen);
                    newIndex.put(command.getKey(), cmdPos);
                }
                start += cmdLen;
            }
            file.seek(file.length());
            HashMap<String, CommandPos> tablePos = tableIndex.get(tableName);
            tablePos.clear();
            tablePos.putAll(newIndex);

            tableIndex.put(tableName,tablePos);

            file.close();

            LoggerUtil.debug(LOGGER, logFormat, "reload index of the table: "+ tableName+ "= " + tablePos.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Override
    public void set(String table,String key, Object value) {
        if (!tableIndex.containsKey(table)){
            initTable(table);
        }
        try {
            SetCommand command = new SetCommand(table,key, value);
            walLogger.log(command);

            HashMap<String, Command> tableCache = tableMemTable.get(table);
            if (tableCache.size() + 1 >= Const.MAX_MEMTABLE_SIZE){
                //内存表达到阀值  加锁并批量写入磁盘
                tableCache.put(key,command);
                tableMemTable.put(table,tableCache);
                flushMemTableToDisk(table);
            }else {
                // 先保存到tableCache
                tableCache.put(key,command);
                tableMemTable.put(table,tableCache);

            }

            // 如果是主节点，同步命令
            //TODO:// 需要异步实现

            if (clusterSyncService != null && clusterSyncService.getSelfNode().getRole() == Role.MASTER) {
                clusterSyncService.asyncSyncCommand(command);
            }

        } catch (Throwable t) {
            recover();
            throw new RuntimeException(t);
        }
    }

    //内存表达到阀值，批量写入磁盘
    public void flushMemTableToDisk(String table) throws InterruptedException {

        //加锁
        if (indexLock.writeLock().tryLock(10, TimeUnit.SECONDS)) {

            try {
                HashMap<String, Command> tableCache = tableMemTable.get(table);
                HashMap<String, CommandPos> tablePos = tableIndex.get(table);
                if (tableCache.isEmpty()) {
                    // 无数据 无需写入
                    return;
                }

                // 遍历所有内存表中的命令，按顺序写入文件
                for (Map.Entry<String, Command> entry : tableCache.entrySet()) {
                    SetCommand command = (SetCommand) entry.getValue();
                    byte[] commandBytes = JSONObject.toJSONBytes(command);


                    //判断key是否已存在
                    CommandPos existingPos = tablePos.get(command.getKey());
                    if (existingPos != null) {
                        //无论是以类型SET还是类型RM存在 先完全删除再更新

                        RandomAccessFileUtil.delteByIndex(this.genFilePath(table), existingPos.getPos(), existingPos.getLen());
                        reloadIndex();

                        // 写table（wal）文件
                        RandomAccessFileUtil.writeInt(this.genFilePath(table), commandBytes.length);
                        int pos = RandomAccessFileUtil.write(this.genFilePath(table), commandBytes);

                        // 添加索引
                        CommandPos cmdPos = new CommandPos(pos, commandBytes.length);
                        tablePos.put(command.getKey(), cmdPos);

                        tableIndex.put(table, tablePos);

                    } else {
                        //不存在 直接进行新增操作

                        // 写table（wal）文件
                        RandomAccessFileUtil.writeInt(this.genFilePath(table), commandBytes.length);
                        int pos = RandomAccessFileUtil.write(this.genFilePath(table), commandBytes);

                        // 添加索引
                        CommandPos cmdPos = new CommandPos(pos, commandBytes.length);
                        tablePos.put(command.getKey(), cmdPos);

                        tableIndex.put(table, tablePos);

                    }
                }
                // 写入完成后，清空内存表
                tableCache.clear();
                tableMemTable.put(table, tableCache);

            } catch (Throwable t) {
                throw new RuntimeException("Failed to flush memTable to disk", t);
            } finally {
                indexLock.writeLock().unlock();
                //刷新缓存
                reloadIndex();
            }
        }else {
            LoggerUtil.error(LOGGER,new RuntimeException(),logFormat,"flushMemTableToDisk超时");
        }
    }

    @Override
    public Object get(String table,String key) {
        if (!tableIndex.containsKey(table)){
            initTable(table);
        }
        try {
            indexLock.readLock().lock();

            // 先读缓存
            HashMap<String, Command> tableCache = tableMemTable.get(table);
            Command cachedCmd = tableCache.get(key);
            if (cachedCmd instanceof SetCommand) {
                // 返还缓存内容
                Object value = ((SetCommand) cachedCmd).getValue();
                if (value instanceof List || value instanceof Map) {
                    return JSONObject.toJSONString(value);
                } else {
                    return String.valueOf(value);
                }
            }
            if (cachedCmd instanceof RmCommand) {
                return null;
            }

            //缓存找不到再找文件
            //从索引中获取信息
            HashMap<String, CommandPos> tablePos = tableIndex.get(table);
            CommandPos cmdPos = tablePos.get(key);
            if (cmdPos == null) {
                return null;
            }
            byte[] commandBytes = RandomAccessFileUtil.readByIndex(this.genFilePath(table), cmdPos.getPos(), cmdPos.getLen());

            JSONObject value = JSONObject.parseObject(new String(commandBytes));
            Command cmd = CommandUtil.jsonToCommand(value);
            if (cmd instanceof SetCommand) {
                tableCache.put(key, cmd);
                tableMemTable.put(table, tableCache);
                return ((SetCommand) cmd).getValue();
            }
            if (cmd instanceof RmCommand) {
                return null;
            }

        } catch (Throwable t) {
            throw new RuntimeException(t);
        } finally {
            indexLock.readLock().unlock();
        }
        return null;
    }

    @Override
    public void rm(String table,String key) {
        if (!tableIndex.containsKey(table)){
            initTable(table);
        }
        try {
            RmCommand command = new RmCommand(table,key);

            byte[] commandBytes = JSONObject.toJSONBytes(command);
            // 加锁
            indexLock.writeLock().lock();

            walLogger.log(command);
            //删除缓存
            HashMap<String, Command> tableCache = tableMemTable.get(table);
            HashMap<String, CommandPos> tablePos = tableIndex.get(table);
            tableCache.remove(key);
            tableMemTable.put(table, tableCache);
            memTable.remove(key);


            CommandPos cmdPos = tablePos.get(key);
            if (cmdPos == null) {
                return;
            }
            RandomAccessFileUtil.removeByIndex(this.genFilePath(table), cmdPos.getPos(), cmdPos.getLen());

            // 删除索引
            tablePos.remove(key);
            tableIndex.put(table, tablePos);

            // 如果是主节点，同步命令
            if (clusterSyncService != null && clusterSyncService.getSelfNode().getRole() == Role.MASTER) {
                clusterSyncService.asyncSyncCommand(command);
            }

        } catch (Throwable t) {
            recover();
            throw new RuntimeException(t);
        } finally {
            indexLock.writeLock().unlock();
            //刷新缓存
            reloadIndex();
        }
    }

    //恢复数据
    public void recover() {
        try {
            List<Command> entries = walLogger.readAll();
            for (Command e : entries) {
                String table = e.getTable();
                memTable.put(e.getKey(), e);
            }
            // 恢复完毕后，清空日志以避免重复重演
            walLogger.clear();
            System.out.println("数据恢复完成，当前数据：" + memTable);
        } catch (IOException e) {
            System.err.println("恢复数据时发生异常：" + e.getMessage());
        }
    }

    // 添加获取节点角色的方法
    public Role getNodeRole() {
        return clusterSyncService != null ?
                clusterSyncService.getSelfNode().getRole() : null;
    }

    public void applyCommand(Command command) {
        System.out.printf("[APPLY] 从节点执行 %s: key=%s%n",
                command.getClass().getSimpleName(), command.getKey());
        try {
            if (command instanceof SetCommand) {
                SetCommand cmd = (SetCommand) command;
                internalSet(cmd.getTable(), cmd.getKey(), cmd.getValue());
            } else if (command instanceof RmCommand) {
                RmCommand cmd = (RmCommand) command;
                internalRm(cmd.getTable(), cmd.getKey());
            }
        } catch (Exception e) {
            LoggerUtil.error(LOGGER, e, logFormat, "应用命令失败");
        }
    }

    private void internalSet(String table, String key, Object value) {
        // 直接写入存储，不触发同步
        SetCommand command = new SetCommand(table, key, value);
        HashMap<String, Command> tableCache = tableMemTable.get(table);
        tableCache.put(key, command);
        tableMemTable.put(table, tableCache);

        if (tableCache.size() + 1 >= Const.MAX_MEMTABLE_SIZE) {
            try {
                flushMemTableToDisk( table);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }finally {
                reloadIndex();
            }
        }
    }

    private void internalRm(String table, String key) {
        // 直接删除，不触发同步
        HashMap<String, Command> tableCache = tableMemTable.get(table);
        HashMap<String, CommandPos> tablePos = tableIndex.get(table);
        tableCache.remove(key);

        CommandPos cmdPos = tablePos.get(key);
        if (cmdPos != null) {
            RandomAccessFileUtil.removeByIndex(this.genFilePath(table), cmdPos.getPos(), cmdPos.getLen());
            tablePos.remove(key);
        }
        tableMemTable.put(table, tableCache);
        tableIndex.put(table, tablePos);
        reloadIndex();
    }

    private void initTable(String table) {
        tableIndex.put(table, new HashMap<String,CommandPos>());
        tableMemTable.put(table, new HashMap<String, Command>());
        List<String> name = new ArrayList<>();
        name.add(table);
        ConfigLoader.writeTableNames(name);
    }

    @Override
    public void close() throws IOException {
        if (memTable != null && !memTable.isEmpty()) {
            LoggerUtil.info(LOGGER, "正在关闭存储，开始持久化 memTable 数据");
            try {
                List<String> tableNames = ConfigLoader.readTableNames();
                for (String tableName : tableNames) {
                    // 将内存表写入磁盘
                    flushMemTableToDisk(tableName);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IOException("MemTable 写入磁盘失败", e);
            }
            LoggerUtil.info(LOGGER, "memTable 数据已持久化");
        }
    }

}
